package com.gjy.rabbitmq.resolve;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 可靠消息消费者
 * 提供消息幂等性处理、手动确认机制和重复消费防护
 * 确保消息不重复消费
 *
 * @author gjy
 * @version 1.0
 * @since 2025-10-13 14:39:09
 */
public class ReliableMessageConsumer {

    private static final Logger logger = LoggerFactory.getLogger(ReliableMessageConsumer.class);

    /**
     * RabbitMQ通道
     */
    private final Channel channel;

    /**
     * 消息处理器
     */
    private final MessageProcessor messageProcessor;

    /**
     * 已处理消息ID集合（内存去重）
     */
    private final Set<String> processedMessageIds = ConcurrentHashMap.newKeySet();

    /**
     * 同步锁
     */
    private final Object lock = new Object();

    /**
     * 构造可靠消息消费者
     *
     * @param channel          RabbitMQ通道
     * @param messageProcessor 消息处理器
     */
    public ReliableMessageConsumer(Channel channel, MessageProcessor messageProcessor) {
        this.channel = channel;
        this.messageProcessor = messageProcessor;
        logger.info("可靠消息消费者初始化完成");
    }

    /**
     * 消费消息（防止重复消费）
     *
     * @param queueName 队列名称
     * @return 消费者标签
     * @throws IOException 消费操作异常
     */
    public String consumeMessages(String queueName) throws IOException {
        // 设置预取消息数量为1，确保消息处理的顺序性和避免内存溢出
        channel.basicQos(1);

        // 创建消息消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                // 获取消息ID和交付标签
                String messageId = properties.getMessageId();
                long deliveryTag = envelope.getDeliveryTag();

                try {
                    logger.debug("接收到消息: messageId={}, deliveryTag={}", messageId, deliveryTag);

                    // 检查消息是否已处理（幂等性保证）
                    if (isMessageProcessed(messageId)) {
                        logger.info("消息已处理，跳过: messageId={}", messageId);
                        channel.basicAck(deliveryTag, false);
                        return;
                    }

                    // 解析消息内容
                    String message = new String(body, StandardCharsets.UTF_8);

                    // 处理消息
                    boolean success = messageProcessor.process(message, properties, envelope);

                    if (success) {
                        // 标记消息为已处理
                        markMessageAsProcessed(messageId);
                        // 确认消息
                        channel.basicAck(deliveryTag, false);
                        logger.info("消息处理成功: messageId={}", messageId);
                    } else {
                        // 拒绝消息并重新入队
                        channel.basicNack(deliveryTag, false, true);
                        logger.warn("消息处理失败，重新入队: messageId={}", messageId);
                    }

                } catch (Exception e) {
                    logger.error("消息处理异常: messageId={}", messageId, e);
                    // 拒绝消息，根据业务需求决定是否重新入队
                    channel.basicNack(deliveryTag, false, true);
                }
            }
        };

        // 开始消费消息
        return channel.basicConsume(queueName, false, consumer);
    }

    /**
     * 检查消息是否已处理
     *
     * @param messageId 消息ID
     * @return 是否已处理
     */
    private boolean isMessageProcessed(String messageId) {
        if (messageId == null) {
            logger.warn("消息ID为空，无法判断是否已处理");
            return false;
        }
        return processedMessageIds.contains(messageId);
    }

    /**
     * 标记消息为已处理
     *
     * @param messageId 消息ID
     */
    private void markMessageAsProcessed(String messageId) {
        if (messageId != null) {
            processedMessageIds.add(messageId);
            // 定期清理已处理的消息ID，避免内存泄漏
            cleanupProcessedMessages();
        }
    }

    /**
     * 清理已处理的消息ID
     * 避免内存泄漏
     */
    private void cleanupProcessedMessages() {
        synchronized (lock) {
            if (processedMessageIds.size() > 10000) {
                logger.info("清理已处理消息ID缓存，当前数量: {}", processedMessageIds.size());
                // 实际应用中可以根据时间戳清理或采用其他策略
            }
        }
    }

}
